package com.primeton.wh.mcenter.impl.mqueue.consumer;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.primeton.wh.mcenter.api.model.Message;
import com.primeton.wh.mcenter.impl.message.service.IMessageService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Calendar;

/**
 * @author: YunTao.Li
 * @create: 2020-03-16 20:45
 * @description:
 **/
@Component
public class KafkaReceiver {
    private Logger log = LoggerFactory.getLogger(KafkaReceiver.class);
    private Gson gson = new GsonBuilder().create();

    @Autowired
    private IMessageService messageService;

//    @KafkaListener(topics = {"topic1"})
    public void receive(ConsumerRecord data) {
        System.out.println("app_log--消费消息:" + data.toString());
        try {

            // step1 : 还原实体对象
            Message record = gson.fromJson(data.value().toString(), Message.class);
            record.setMcSendtime(Calendar.getInstance().getTime());

            // step2 : 持久化
            messageService.addMessage(record); // TODO:异步
        } catch (Throwable throwable) {
            throwable.printStackTrace();
        }

    }
}
